package drds.server.server;

import java.nio.ByteBuffer;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import drds.server.config.ErrorCode;
import drds.server.datasource.IConnectionHolder;
import drds.server.handler.CommitNodeHandler;
import drds.server.handler.MultiNodeQueryHandler;
import drds.server.handler.RollbackNodeHandler;
import drds.server.handler.RollbackReleaseHandler;
import drds.server.handler.SingleNodeHandler;
import drds.server.net.packet.OkPacket;
import drds.server.route.Route;
import drds.server.route.RouteNode;
import drds.server.server.parser.ServerParse;

public class Session implements ISession {

	public static final Logger LOGGER = LoggerFactory.getLogger(Session.class);

	private final SessionContext sessionContext;
	private final ConcurrentHashMap<RouteNode, IConnectionHolder> nodeConnectionHolderMap;

	private volatile SingleNodeHandler singleNodeHandler;
	private volatile MultiNodeQueryHandler multiNodeQueryHandler;

	private volatile RollbackNodeHandler rollbackHandler;
	private final CommitNodeHandler commitHandler;

	private boolean prepared;

	public Session(SessionContext sessionContext) {
		this.sessionContext = sessionContext;
		this.nodeConnectionHolderMap = new ConcurrentHashMap<RouteNode, IConnectionHolder>(2, 0.75f);
		commitHandler = new CommitNodeHandler(this);
	}

	@Override
	public SessionContext getSessionContext() {
		return sessionContext;
	}

	@Override
	public int getNodeSize() {
		return nodeConnectionHolderMap.size();
	}

	public Set<RouteNode> getNodeSet() {
		return nodeConnectionHolderMap.keySet();
	}

	public IConnectionHolder getConnectionHolder(RouteNode node) {
		return nodeConnectionHolderMap.get(node);
	}

	public Map<RouteNode, IConnectionHolder> getRouteNodeWithConnectionHolderMap() {
		return this.nodeConnectionHolderMap;
	}

	public IConnectionHolder removeTarget(RouteNode node) {
		return nodeConnectionHolderMap.remove(node);
	}

	@Override
	public void execute(Route route, int type) {

		// clear prev execute resources
		clearHandlesResources();

		// 检查路由结果是否为空
		RouteNode[] routeNodes = route.getRouteNodes();
		if (routeNodes == null || routeNodes.length == 0 || routeNodes[0].getName() == null || routeNodes[0].getName().equals("")) {
			sessionContext.writeErrorMessage(ErrorCode.ER_NO_DB_ERROR, "No dataNode found ,please check tables defined in schema");
			return;
		}
		boolean autoCommit = sessionContext.isAutoCommit();
		final int initCount = nodeConnectionHolderMap.size();
		if (routeNodes.length == 1) {
			singleNodeHandler = new SingleNodeHandler(route, this);
			if (this.isPrepared()) {
				singleNodeHandler.setPrepared(true);
			}

			try {
				if (initCount > 1) {
					checkDistriTransaxAndExecute(route, 1, autoCommit);
				} else {
					singleNodeHandler.execute();
				}
			} catch (Exception e) {
				LOGGER.error(new StringBuilder().append(sessionContext).append(route).toString(), e);
				sessionContext.writeErrorMessage(ErrorCode.ERR_HANDLE_DATA, e.toString());
			}

		} else {

			multiNodeQueryHandler = new MultiNodeQueryHandler(type, route, autoCommit, this);
			if (this.isPrepared()) {
				multiNodeQueryHandler.setPrepared(true);
			}
			try {
				if (((type == ServerParse.DELETE || type == ServerParse.INSERT || type == ServerParse.UPDATE) && !route.isGlobalTable() && routeNodes.length > 1) || initCount > 1) {
					checkDistriTransaxAndExecute(route, 2, autoCommit);
				} else {
					multiNodeQueryHandler.execute();
				}

			} catch (Exception e) {
				LOGGER.error(new StringBuilder().append(sessionContext).append(route).toString(), e);
				sessionContext.writeErrorMessage(ErrorCode.ERR_HANDLE_DATA, e.toString());
			}
		}

		if (this.isPrepared()) {
			this.setPrepared(false);
		}
	}

	private void checkDistriTransaxAndExecute(Route route, int type, boolean autoCommit) throws Exception {
		if (type == 1) {
			singleNodeHandler.execute();
		} else {
			multiNodeQueryHandler.execute();
		}
	}

	public void commit() {
		final int initCount = nodeConnectionHolderMap.size();
		if (initCount <= 0) {
			ByteBuffer byteBuffer = sessionContext.allocate();
			byteBuffer = sessionContext.writeToByteBuffer(OkPacket.OK, byteBuffer);
			sessionContext.write(byteBuffer);
			return;
		} else if (initCount == 1) {
			IConnectionHolder connectionHolder = nodeConnectionHolderMap.elements().nextElement();
			commitHandler.commit(connectionHolder);

		} else {

		}

	}

	public void rollBack() {
		final int initCount = nodeConnectionHolderMap.size();
		if (initCount <= 0) {
			ByteBuffer byteBuffer = sessionContext.allocate();
			byteBuffer = sessionContext.writeToByteBuffer(OkPacket.OK, byteBuffer);
			sessionContext.write(byteBuffer);
			return;
		}

		rollbackHandler = new RollbackNodeHandler(this);
		rollbackHandler.rollback();
	}

	@Override
	public void cancel(SessionContext sessionContext) {

	}

	/**
	 * {@link SessionContext#isClosed()} must be true before invoking this
	 */
	public void terminate() {
		for (IConnectionHolder connectionHolder : nodeConnectionHolderMap.values()) {
			connectionHolder.close("client closed ");
		}
		nodeConnectionHolderMap.clear();
		clearHandlesResources();
	}

	/**
	 * 对session里面所有的连接进行关闭
	 */
	public void closeAndClearResources(String reason) {
		for (IConnectionHolder connectionHolder : nodeConnectionHolderMap.values()) {
			connectionHolder.close(reason);
		}
		nodeConnectionHolderMap.clear();
		clearHandlesResources();
	}

	public void releaseConnectionIfSafe(IConnectionHolder connectionHolder, boolean needRollback) {
		RouteNode node = (RouteNode) connectionHolder.getAttachment();

		if (node != null) {
			if (node.isDisctTable()) {
				return;
			}
			if (this.sessionContext.isAutoCommit() || connectionHolder.isSlaveDataBase() || !connectionHolder.isModifiedSQLExecuted()) {
				releaseConnection((RouteNode) connectionHolder.getAttachment(), needRollback);
			}
		}
	}

	public void releaseConnection(RouteNode node, final boolean needRollback) {

		IConnectionHolder connectionHolder = nodeConnectionHolderMap.remove(node);
		if (connectionHolder != null) {
			if (connectionHolder.getAttachment() != null) {
				connectionHolder.setAttachment(null);
			}
			if (!connectionHolder.isClosedOrQuit()) {
				if (connectionHolder.isAutoCommit()) {
					connectionHolder.release();
				} else
				// if (needRollback)
				{
					connectionHolder.setHandler(new RollbackReleaseHandler());
					connectionHolder.rollback();
				}
				// else {
				// c.release();
				// }
			}
		}
	}

	public void releaseConnections(final boolean needRollback) {
		for (RouteNode node : nodeConnectionHolderMap.keySet()) {
			releaseConnection(node, needRollback);
		}
	}

	public void releaseConnection(IConnectionHolder connectionHolder) {
		Iterator<Entry<RouteNode, IConnectionHolder>> iterator = nodeConnectionHolderMap.entrySet().iterator();
		while (iterator.hasNext()) {
			IConnectionHolder $connectionHolder = iterator.next().getValue();
			if ($connectionHolder == connectionHolder) {
				iterator.remove();
				connectionHolder.release();
				break;
			}
		}

	}

	/**
	 * @return previous bound connection
	 */
	public IConnectionHolder bind(RouteNode node, IConnectionHolder connectionHolder) {
		return nodeConnectionHolderMap.put(node, connectionHolder);
	}

	public boolean tryExistsCon(final IConnectionHolder connectionHolder, RouteNode node) {
		if (connectionHolder == null) {
			return false;
		}

		boolean canReUse = false;
		// conn 是 slave db 的，并且 路由结果显示，本次sql可以重用该 conn
		if (connectionHolder.isSlaveDataBase() && (node.canRunnOnReadDataBase(getSessionContext().isAutoCommit()) && (node.masterTrue$SlaveFalse() == null || node.masterTrue$SlaveFalse()))) {
			canReUse = true;
		}

		// conn 是 master db 的，并且路由结果显示，本次sql可以重用该conn
		if (!connectionHolder.isSlaveDataBase() && (node.masterTrue$SlaveFalse() == null || !node.masterTrue$SlaveFalse())) {
			canReUse = true;
		}

		if (canReUse) {
			connectionHolder.setAttachment(node);
			return true;
		} else {
			// slavedb connection and can't use anymore ,release it
			releaseConnection(node, false);
		}
		return false;
	}

	/**
	 * KillConnection 服务器端线程ID 同时回收数据库连接 回滚事物 连接池连接归还。
	 */
	protected void kill() {

	}

	private void clearHandlesResources() {
		SingleNodeHandler singleNodeHandler = this.singleNodeHandler;
		if (singleNodeHandler != null) {
			singleNodeHandler.clearResources();
			this.singleNodeHandler = null;
		}
		MultiNodeQueryHandler multiNodeQueryHandler = this.multiNodeQueryHandler;
		if (multiNodeQueryHandler != null) {
			multiNodeQueryHandler.clearResources();
			this.multiNodeQueryHandler = null;
		}
	}

	public void clearResources(final boolean needRollback) {
		this.releaseConnections(needRollback);
		clearHandlesResources();
	}

	public boolean closed() {
		return sessionContext.isClosed();
	}

	public boolean isPrepared() {
		return prepared;
	}

	public void setPrepared(boolean prepared) {
		this.prepared = prepared;
	}

}